package edu.del.newsreleasesystem.mq;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;
import edu.del.newsreleasesystem.config.RabbitMQConfig;
import edu.del.newsreleasesystem.es.service.NewsEsIndexService;
import edu.del.newsreleasesystem.mq.message.NewsSyncMessage;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Map;

@Slf4j
@Component
@RequiredArgsConstructor
public class NewsSyncConsumer {

    private final ObjectMapper objectMapper = new ObjectMapper();

    private final NewsEsIndexService newsEsIndexService;

    /**
     * 监听同步队列，手动 ack
     *
     * @param message  消息体，包含 NewsSyncMessage 的 JSON 数据
     * @param channel  RabbitMQ Channel，用于手动确认
     * @param headers  消息头（可获取 deliveryTag 等）
     */
    @RabbitListener(queues = RabbitMQConfig.SYNC_QUEUE)
    public void consumeNewsSyncMessage(
            Message message,
            @Headers Map<String, Object> headers,
            Channel  channel
            ) {
        // 获取消息的唯一标识，用于 ack/nack
        long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
        try {
            // 1. 提取消息体字节数组
            byte[] body = message.getBody();
            if (body == null || body.length == 0) {
                log.warn("收到空消息，deliveryTag: {}", deliveryTag);
                // 拒绝消息并丢弃（不重新入队）
                channel.basicNack(deliveryTag, false, false);
                return;
            }

            // 2. 反序列化为 NewsSyncMessage 对象
            NewsSyncMessage newsSyncMessage = objectMapper.readValue(body, NewsSyncMessage.class);
            log.info("消费者收到新闻同步消息：newsId={}, 操作类型={}, 事件时间={}",
                    newsSyncMessage.getNewsId(),
                    newsSyncMessage.getOperationType(),
                    newsSyncMessage.getEventTime());

            // 3. 根据不同的操作类型执行同步逻辑
            handleNewsSyncOperation(newsSyncMessage);

            // 4. 消息处理成功，手动确认（ack）
            channel.basicAck(deliveryTag, false);
            log.info("消息处理成功并已确认，deliveryTag: {}", deliveryTag);
        } catch (JsonProcessingException e) {
            log.error("JSON 反序列化失败，消息体可能不是有效的 NewsSyncMessage，deliveryTag: {}", deliveryTag, e);
            // 可选择记录错误消息，或转存死信队列，这里我们拒绝并重新入队
            try {
                // 重新入队
                channel.basicNack(deliveryTag, false, true);
            } catch (IOException ioException) {
                log.error("Nack 失败，deliveryTag: {}", deliveryTag, ioException);
            }
        }catch (Exception e) {
            log.error("处理新闻同步消息时发生未知异常，deliveryTag: {}", deliveryTag, e);
            // 出现未知异常，也选择重新入队或进入死信队列
            try {
                // 这里选择重新入队
                channel.basicNack(deliveryTag, false, true);
            } catch (IOException ioException) {
                log.error("Nack 失败，deliveryTag: {}", deliveryTag, ioException);
            }
        }
    }

    /**
     * 根据操作类型执行具体的同步逻辑
     *
     * @param newsSyncMessage 同步消息
     */
    private void handleNewsSyncOperation(NewsSyncMessage newsSyncMessage) {
        NewsSyncMessage.OperationType operationType = newsSyncMessage.getOperationType();
        Integer  newsId= newsSyncMessage.getNewsId();
        try {
            switch (operationType) {
                case PUBLISH:
                case UPDATE:
                    newsEsIndexService.syncNewsToEsOnPublishOrUpdate(newsId);
                    break;
                case DELETE:
                    newsEsIndexService.syncNewsToEsOnDelete(newsId);
                    break;
                default:
                    log.warn("未知的操作类型: {}", operationType);
            }
        } catch (Exception e) {
            log.error("执行同步操作时出错，新闻ID: {}, 操作类型: {}", newsId, operationType, e);
        }
    }
}
